package com.ds.infra.mq.core.network;

import com.ds.infra.mq.core.constants.MQConstant;
import com.ds.infra.mq.core.protocol.Message;
import com.ds.infra.mq.core.protocol.MessageHeader;
import com.ds.infra.mq.core.protocol.MessageType;
import com.ds.infra.mq.core.storage.MemoryMessageStore;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * 处理客户端请求的处理器，包括生产者和消费者的请求。
 *
 * @author WeiShaoying
 */
@Slf4j
public class RequestProcessor {

    private final MemoryMessageStore messageStore;
    private final long maxWaitTimeMs = 3000; // 最大等待时间(毫秒)
    private final int retryIntervalMs = 100;  // 重试间隔(毫秒)

    public RequestProcessor(MemoryMessageStore messageStore) {
        this.messageStore = messageStore;
    }

    /**
     * 处理生产者请求
     *
     * @param request 请求消息
     * @return 处理结果消息
     */
    public Message handleProduceRequest(Message request) {
        try {
            // 1. 解析请求体 (格式: topic[:partition]|messageBody)
            String requestBody = new String(request.getBody(), StandardCharsets.UTF_8);
            String[] parts = parseProduceRequest(requestBody);
            String topic = parts[0];
            int partition = Integer.parseInt(parts[1]);
            String messageBody = parts[2];

            // 2. 存储消息
            Message messageToStore = new Message();
            messageToStore.setBody(messageBody.getBytes(StandardCharsets.UTF_8));
            messageStore.putMessage(topic, partition, messageToStore);

            // 3. 获取写入偏移量
            long offset = messageStore.getWriteOffset(topic, partition);

            return buildSuccessResponse(
                    request,
                    MessageType.PRODUCE_RESPONSE,
                    ("offset:" + offset).getBytes(StandardCharsets.UTF_8)
            );
        } catch (IllegalArgumentException e) {
            return buildErrorResponse(request, "Invalid request format: " + e.getMessage());
        } catch (Exception e) {
            log.error("Produce request failed", e);
            return buildErrorResponse(request, "Produce failed: " + e.getMessage());
        }
    }

    /**
     * 处理消费者请求
     *
     * @param request 请求消息
     * @return 处理结果消息
     */
    public Message handleConsumeRequest(Message request) {
        try {
            // 1. 解析请求体 (格式: "topic[:partition]")
            String requestBody = new String(request.getBody(), StandardCharsets.UTF_8);
            String[] parts = parseConsumeRequest(requestBody);
            String topic = parts[0];
            int partition = Integer.parseInt(parts[1]);
            String consumerId = request.getHeader().getClientId();

            // 2. 等待消息到达
            Message message = waitForMessage(topic, partition, consumerId);

            // 3. 构建响应
            if (message != null) {
                log.debug("Delivered message to {} [topic:{}, partition:{}]", consumerId, topic, partition);
                return buildSuccessResponse(
                        request,
                        MessageType.CONSUME_RESPONSE,
                        message.getBody()
                );
            } else {
                return buildEmptyResponse(request);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return buildErrorResponse(request, "Consume interrupted");
        } catch (Exception e) {
            log.error("Consume request failed", e);
            return buildErrorResponse(request, "Consume failed: " + e.getMessage());
        }
    }

    /**
     * 解析生产请求体，格式为 "topic:partition|messageBody"
     *
     * @param requestBody 请求体字符串
     * @return String[]
     */
    private String[] parseProduceRequest(String requestBody) {
        String[] parts = requestBody.split("\\|", 2);
        if (parts.length != 2 || parts[0].isEmpty() || parts[1].isEmpty()) {
            throw new IllegalArgumentException("Expected 'topic[:partition]|messageBody'");
        }

        String[] topicParts = parts[0].split(":");
        String topic = topicParts[0];
        String partition = topicParts.length > 1 ? topicParts[1] : "0"; // 默认分区0

        return new String[]{topic, partition, parts[1]};
    }

    private String[] parseConsumeRequest(String requestBody) {
        String[] parts = requestBody.split(":");
        String topic = parts[0];
        String partition = parts.length > 1 ? parts[1] : "0"; // 默认分区0
        return new String[]{topic, partition};
    }

    private Message waitForMessage(String topic, int partition, String consumerId)
            throws InterruptedException {
        long startTime = System.currentTimeMillis();
        Message message = null;
        // 等待直到超时或获取到消息
        while (message == null && (System.currentTimeMillis() - startTime) < maxWaitTimeMs) {
            message = messageStore.getMessage(topic, partition, consumerId);
            if (message == null) {
                if (!messageStore.topicExists(topic)) {
                    break; // 主题不存在则立即返回
                }
                // 等待一段时间后重试
                TimeUnit.MILLISECONDS.sleep(retryIntervalMs);
            }
        }
        return message;
    }

    private Message buildSuccessResponse(Message request, MessageType type, byte[] body) {
        Message response = new Message();
        response.setHeader(createHeader(request, type));
        response.setBody(body);
        return response;
    }

    private Message buildErrorResponse(Message request, String errorMsg) {
        Message response = new Message();
        response.setHeader(createHeader(request, MessageType.PRODUCE_RESPONSE));
        response.setBody(("ERROR:" + errorMsg).getBytes(StandardCharsets.UTF_8));
        return response;
    }

    private Message buildEmptyResponse(Message request) {
        Message response = new Message();
        MessageHeader header = new MessageHeader();
        header.setMessageType(MessageType.CONSUME_RESPONSE.getCode());
        header.setRequestId(request.getHeader().getRequestId());
        header.setClientId(request.getHeader().getClientId());
        response.setHeader(header);
        response.setBody(MQConstant.EMPTY_STR.getBytes(StandardCharsets.UTF_8)); // 明确返回EMPTY
        return response;
    }

    private MessageHeader createHeader(Message request, MessageType type) {
        MessageHeader header = new MessageHeader();
        header.setMessageType(type.getCode());
        header.setRequestId(request.getHeader().getRequestId());
        header.setClientId(request.getHeader().getClientId());
        return header;
    }

    public Message handleHeartbeat(Message request) {
        return buildSuccessResponse(
                request,
                MessageType.HEARTBEAT,
                "OK".getBytes(StandardCharsets.UTF_8)
        );
    }
}
